package com.hgl.bi.web.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hgl.bi.common.config.COSClientConfig;
import com.hgl.bi.common.constant.ErrorCode;
import com.hgl.bi.common.constant.ProgressStatus;
import com.hgl.bi.common.exception.BusinessException;
import com.hgl.bi.common.manager.COSManager;
import com.hgl.bi.common.utils.AssertUtil;
import com.hgl.bi.common.utils.FileUtil;
import com.hgl.bi.web.config.DeepSeekClient;
import com.hgl.bi.web.config.RabbitConfig;
import com.hgl.bi.web.entity.AnalyzeEntity;
import com.hgl.bi.web.entity.SetEntity;
import com.hgl.bi.web.mapper.SetMapper;
import com.hgl.bi.web.model.dto.AnalMsgDto;
import com.hgl.bi.web.service.AnalyzeService;
import com.hgl.bi.web.service.SetService;
import com.hgl.bi.web.service.UserService;
import com.qcloud.cos.COSClient;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;

/**
 * @program: BI
 * @description: 数据清洗消费者
 * @author: hgl
 * @create: 2025-02-15 23:17
 */
@Slf4j
@Component
public class AnalyzeConsumer {

    @Autowired
    private AnalyzeService analyzeService;
    
    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private DeepSeekClient deepSeekClient;

    @Autowired
    private COSManager cosManager;

    @Autowired
    private SetService setService;

    @Autowired
    private SetMapper setMapper;

    @Autowired
    private COSClient cosClient;

    @Autowired
    private COSClientConfig cosClientConfig;

    @Autowired
    private UserService userService;

    @RabbitListener(queues = RabbitConfig.DATA_ANALYZE_QUEUE)
    public void doDataAnalyze(@Payload AnalMsgDto messageDto, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        // 解析消息
        Long analyzeId = messageDto.getAnalyzeId();
        Long setId = messageDto.getSetId();
        String demand = messageDto.getDemand();
        // 非空校验
        AssertUtil.notNull(analyzeId, ErrorCode.ANALYZE_ID_NULL);
        AssertUtil.notNull(setId, ErrorCode.SET_ID_NULL);
        AssertUtil.notEmpty(demand, ErrorCode.DEMAND_NOT_NULL);

        try {
            // 1. 获取待处理任务（使用乐观锁）
            AnalyzeEntity analyze = analyzeService.lambdaQuery()
                    .eq(AnalyzeEntity::getId, analyzeId)
                    .eq(AnalyzeEntity::getProgress, ProgressStatus.WAITING.getCode())
                    .oneOpt()
                    .orElseThrow(() -> new BusinessException(ErrorCode.TASK_NOT_EXIST_OR_PROCESSED));
            // 2. 更新为处理中状态
            analyze.setProgress(ProgressStatus.RUNNING.getCode());
            analyze.setStartTime(new Date());
            if (!analyzeService.updateById(analyze)) {
                throw new BusinessException(ErrorCode.STATUS_UPDATE_ERROR);
            }
            // 3. 执行业务逻辑
            processAnalyze(analyze, setId, demand);
            // 4. 确认消息
            channel.basicAck(tag, false);
        } catch (BusinessException e) {
            // 业务异常不重试
            log.error("任务处理失败 taskId:{}", analyzeId, e);
            basicNackWithRetry(channel, tag); // 最大重试3次
        } catch (Exception e) {
            log.error("系统异常 taskId:{}", analyzeId, e);
            basicNackWithRetry(channel, tag);
        }
    }

    private void processAnalyze(AnalyzeEntity analyze, Long setId, String demand) {
        try {
            // 获取cos path
            ArrayList<String> paths = setMapper.selectPathBySetId(setId);
            StringBuilder csvContent = new StringBuilder();
            int i = 1;
            // 读取文件内容
            for (String path : paths) {
                csvContent.append("数据集文件").append(i).append("：\n");
                List<Map<Integer, String>> content = cosManager.syncReadContent(path);
                // 转换为CSV格式
                String csv = FileUtil.toCsv(content);
                csvContent.append(csv);
                csvContent.append(",\n");
                i++;
            }
            csvContent.append("需求：").append(demand);
            log.info("CSV数据: {}", csvContent);
            // 调用DeepSeek API
            String prompt = FileUtil.loadPromptText(false);
            log.info("提示词: {}", prompt);
            log.info("正在调用{}模型执行分析任务: {}",deepSeekClient.getModel(), analyze.getId());
            String result = deepSeekClient.callDeepSeekApi(prompt, String.valueOf(csvContent));
            // 解析JSON，映射对象
            AssertUtil.notEmpty(result, ErrorCode.RESPONSE_IS_NULL);
            String[] split = result.split("【【【");
            AssertUtil.isTrue(!(split.length < 3), ErrorCode.RESPONSE_ERROR);
            String vizConfig = split[1].trim();
            String conclusion = split[2].trim();
            // 更新任务状态
            analyze.setEndTime(new Date());
            analyze.setVizConfig(vizConfig);
            analyze.setConclusion(conclusion);
            analyze.setProgress(ProgressStatus.SUCCESS.getCode());
            analyzeService.updateById(analyze);
            // 更新数据集分析次数
            SetEntity set = setService.getById(setId);
            set.setAnalyzeCount(set.getAnalyzeCount() + 1);
            setService.updateById(set);
        } catch (Exception e) {
            // 记录详细错误日志
            analyze.setEndTime(new Date());
            analyze.setErrorLog(ExceptionUtils.getMessage(e));
            analyze.setProgress(ProgressStatus.FAILURE.getCode());
            analyzeService.updateById(analyze);
        }
    }

    private void basicNackWithRetry(Channel channel, long tag) {
        try {
            GetResponse response = channel.basicGet(RabbitConfig.DATA_ANALYZE_QUEUE, false);
            if (response == null) {
                channel.basicNack(tag, false, false);
                return;
            }

            AMQP.BasicProperties props = response.getProps();
            Integer retryCount = (Integer) props.getHeaders().getOrDefault("x-retry-count", 0);

            if (retryCount < 3) {
                // 重新发布消息
                channel.basicPublish("", RabbitConfig.DATA_ANALYZE_QUEUE,
                        new AMQP.BasicProperties.Builder()
                                .headers(createHeaders(retryCount))
                                .build(),
                        response.getBody());

                channel.basicAck(tag, false);
            } else {
                channel.basicNack(tag, false, false); // 进入死信队列
            }
        } catch (IOException ex) {
            log.error("消息NACK处理失败", ex);
        }
    }

    // Java 8兼容写法替代Map.of()
    private Map<String, Object> createHeaders(int retryCount) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("x-retry-count", retryCount + 1);
        return headers;
    }
}
